package org.apache.flink.connector.elasticsearch.sink;


import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.ssl.SSLContextBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import java.security.cert.X509Certificate;
import java.util.List;

import static org.apache.flink.util.Preconditions.checkState;

/** NetworkConfigConfig A factory that creates valid ElasticsearchClient instances. */
public class NetworkConfig {
    private final List<HttpHost> hosts;

    private final List<Header> headers;

    private final String username;

    private final String password;

    transient CredentialsProvider credentialsProvider;

    public NetworkConfig(
            List<HttpHost> hosts,
            String username,
            String password,
            List<Header> headers,
            String certificateFingerprint) {
        checkState(hosts.size() > 0, "Hosts must not be null");
        this.hosts = hosts;
        this.username = username;
        this.password = password;
        this.headers = headers;
    }

    public ElasticsearchAsyncClient create() {
        return new ElasticsearchAsyncClient(
                new RestClientTransport(this.getRestClient(), new JacksonJsonpMapper()));
    }

    private RestClient getRestClient() {
        if (credentialsProvider == null) {
            credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(
                    AuthScope.ANY,
                    new UsernamePasswordCredentials(username, password));
        }
        RestClientBuilder restClientBuilder =
                RestClient
                        .builder(hosts.toArray(new HttpHost[0]))
                        .setHttpClientConfigCallback(
                                httpAsyncClientBuilder -> {
                                    try {
                                        SSLContextBuilder sslContextBuilder = SSLContextBuilder.create();
                                        sslContextBuilder.loadTrustMaterial(new TrustSelfSignedStrategy() {
                                            @Override
                                            public boolean isTrusted(
                                                    X509Certificate[] chain,
                                                    String authType) {
                                                return true; // 忽略所有证书验证
                                            }
                                        });
                                        httpAsyncClientBuilder.setSSLContext(sslContextBuilder.build());
                                        httpAsyncClientBuilder.setSSLHostnameVerifier(
                                                NoopHostnameVerifier.INSTANCE);
                                    } catch (Exception e) {
                                        throw new RuntimeException(e);
                                    }
                                    return httpAsyncClientBuilder.setDefaultCredentialsProvider(
                                            credentialsProvider);
                                })
                        .setRequestConfigCallback(builder -> {
                            builder.setConnectTimeout(5000);
                            builder.setSocketTimeout(60000);
                            builder.setConnectionRequestTimeout(2000);
                            return builder;
                        });
        ;

        if (headers != null) {
            restClientBuilder.setDefaultHeaders(headers.toArray(new Header[0]));
        }

        return restClientBuilder.build();
    }

    private CredentialsProvider getCredentials() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

        credentialsProvider.setCredentials(
                AuthScope.ANY, new UsernamePasswordCredentials(username, password));

        return credentialsProvider;
    }
}
